Skip to content

Firestore sink - #377

Open
alex-thc wants to merge 3 commits into
mainfrom
firestore
Open

Firestore sink#377
alex-thc wants to merge 3 commits into
mainfrom
firestore

Conversation

@alex-thc

@alex-thc alex-thc commented Apr 2, 2026

Copy link
Copy Markdown
Contributor

Summary by CodeRabbit

  • New Features

    • Added Firestore connector for sink-only operations with batched writes (default batch size 500)
    • Supports Mongo BSON and JSON payloads with deterministic overwrite and delete semantics
    • CLI flags to configure credentials file, batch size, and connector ID
  • Tests

    • Integration and unit tests covering batching, overwrite/delete behavior, and BSON type conversions

@coderabbitai

coderabbitai Bot commented Apr 2, 2026

Copy link
Copy Markdown
📝 Walkthrough

Walkthrough

Added a new Firestore connector with URI parsing, batched sink write operations supporting BSON and JSON type conversion, comprehensive unit and emulator integration tests, CLI registration and flags, and updated module dependencies to support Firestore client libraries.

Changes

Cohort / File(s) Summary
Firestore Connector Core
connectors/firestore/connector.go
New Firestore connector implementation: parses firestore://project[/database], defines ConnectorSettings, NewConn, DefaultBatchSize, validation errors, write paths (WriteData, WriteUpdates) using Firestore BulkWriter, BSON→Firestore conversions, and Teardown. Read/stream methods return unimplemented errors.
Unit Tests & Conversion Tests
connectors/firestore/connector_test.go, connectors/firestore/bson_conversion_test.go
Unit tests for URI parsing, namespace→collection mapping, value stringification, and BSON type conversions (convertBsonTypes, rawToMap) covering ObjectID, DateTime, Binary, Decimal128, arrays, and nested docs.
Integration Tests (Emulator)
connectors/firestore/connector_integration_test.go
New build-tagged emulator tests: full connector test suite (sink-only), batch-write behavior, deterministic overwrite semantics, delete via WriteUpdates, and complex BSON-type roundtrip verification using Firestore emulator.
CLI Registration & Flags
internal/app/options/connectorflags.go
Registered Firestore connector in GetRegisteredConnectors() for firestore:// URIs and added FirestoreFlags(settings *firestore.ConnectorSettings) exposing --credentials-file, --batch-size, and --id.
Dependency Updates
go.mod
Updated dependencies to include Google Cloud Firestore and related libraries, bumped versions for gRPC, protobuf, golang.org/x/time, x/sync, x/oauth2, and added indirect modules required by Firestore client and OpenTelemetry packages.

Sequence Diagram

sequenceDiagram
    participant Client
    participant Connector as Firestore Connector
    participant BulkWriter as Bulk Writer
    participant Firestore as Firestore Service

    Client->>Connector: WriteData(items)
    Connector->>Connector: Parse URI, validate settings
    Connector->>Connector: Extract document IDs, convert BSON/JSON
    Connector->>BulkWriter: Add Set/Delete operations (batching)
    BulkWriter->>Firestore: Commit batch
    Firestore-->>BulkWriter: Commit result
    BulkWriter-->>Connector: Job result / errors
    Connector-->>Client: Return success/error
Loading

Estimated code review effort

🎯 4 (Complex) | ⏱️ ~45 minutes

Possibly related PRs

  • Fakesource #292 — Similar connector registration change to GetRegisteredConnectors(); pattern matches Firestore registration here.
  • S3json-take2 #354 — Adds connector registration and CLI flag helpers (similar structure to Firestore flags).
  • File connector (CSV only for now) #367 — Adds a new connector implementation and registration, following the same connector wiring pattern.

Suggested reviewers

  • adiom-mark

Poem

🐰 I hopped through URIs and mapped each dot,

Batching writes in fields where documents plot,
BSON turned gentle, JSON hummed a tune,
Emulator nights beneath a testing moon,
Now Firestore blooms — a carrot-shaped commit!

🚥 Pre-merge checks | ✅ 2 | ❌ 1

❌ Failed checks (1 warning)

Check name Status Explanation Resolution
Docstring Coverage ⚠️ Warning Docstring coverage is 24.14% which is insufficient. The required threshold is 80.00%. Write docstrings for the functions missing them to satisfy the coverage threshold.
✅ Passed checks (2 passed)
Check name Status Explanation
Description Check ✅ Passed Check skipped - CodeRabbit’s high-level summary is enabled.
Title check ✅ Passed The title 'Firestore sink' directly identifies the primary feature added: a new Firestore connector implementation for writing data. The changeset introduces connector code, tests, and CLI integration for Firestore sink capabilities.

✏️ Tip: You can configure your own custom pre-merge checks in the settings.

✨ Finishing Touches
📝 Generate docstrings
  • Create stacked PR
  • Commit on current branch
🧪 Generate unit tests (beta)
  • Create PR with unit tests
  • Commit unit tests in branch firestore

Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out.

❤️ Share

Comment @coderabbitai help to get the list of available commands and usage tips.

@coderabbitai coderabbitai Bot left a comment

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Actionable comments posted: 3

🧹 Nitpick comments (4)
connectors/firestore/connector.go (3)

91-93: BatchSize validation silently clamps invalid values.

When BatchSize > DefaultBatchSize, it's silently reset to the default. Consider logging a warning so users know their configured value was adjusted.

Proposed improvement
 	if settings.BatchSize <= 0 || settings.BatchSize > DefaultBatchSize {
+		if settings.BatchSize > DefaultBatchSize {
+			slog.Warn("batch size exceeds Firestore limit, clamping to max", "requested", settings.BatchSize, "max", DefaultBatchSize)
+		}
 		settings.BatchSize = DefaultBatchSize
 	}
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.

In `@connectors/firestore/connector.go` around lines 91 - 93, The code silently
clamps invalid settings.BatchSize to DefaultBatchSize; update the if block that
checks settings.BatchSize to emit a warning before changing the value so users
know their configured value was adjusted—e.g., log a message referencing the
original settings.BatchSize and DefaultBatchSize (use the connector's existing
logger, e.g., logger.Warnf or log.Printf) then set settings.BatchSize =
DefaultBatchSize. Ensure you reference settings.BatchSize and DefaultBatchSize
in the warning so it's clear which values were changed.

322-325: Potential precision loss when formatting large uint64 values.

For uint64 values exceeding JavaScript's safe integer limit (2^53 - 1), the %d format will preserve the value, but downstream JSON consumers may lose precision. This is a known limitation when using numeric document IDs with Firestore and JSON.

🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.

In `@connectors/firestore/connector.go` around lines 322 - 325, The current switch
cases for numeric types (case int/int32/int64 and case uint/uint32/uint64) can
cause precision loss for values > JS_SAFE_INT (2^53-1); update the handling so
uint64 (and unsigned types promoted to uint64) are detected and for values >
9007199254740991 you return their decimal representation as a string (use
strconv.FormatUint(val, 10)), otherwise return the numeric formatting;
specifically modify the uint64 handling branch (and any code paths that cast to
uint64) to compare against the JS max safe integer and return a string for large
values to avoid downstream JSON precision loss.

199-220: Batch write lacks error context for identifying problematic documents.

When extractDocumentIDAndData fails, the error doesn't include which document in the batch caused the failure. This makes debugging harder when processing large batches.

Proposed improvement
 	for _, raw := range data {
 		docID, docData, err := extractDocumentIDAndData(raw, dataType)
 		if err != nil {
-			return fmt.Errorf("failed to extract document ID: %w", err)
+			return fmt.Errorf("failed to extract document ID (batch index may help identify doc): %w", err)
 		}

Alternatively, consider adding the document index to the error message.

🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.

In `@connectors/firestore/connector.go` around lines 199 - 220, In writeBatch,
include context about which item failed by changing the loop to capture the item
index and include it in the returned error when extractDocumentIDAndData fails;
update the error returned from extractDocumentIDAndData inside writeBatch (and
any subsequent errors) to wrap the original error with fmt.Errorf including the
index (and optionally a short snippet or hex of raw) and the collectionName so
it's straightforward to identify the problematic document when debugging;
reference conn.writeBatch and extractDocumentIDAndData to locate and modify the
error wrapping logic.
connectors/firestore/connector_test.go (1)

104-125: Consider adding error case and BSON type coverage for valueToString.

The test covers basic scalar types but misses:

  • bson.ObjectID conversion (returns Hex())
  • bson.Binary conversion (returns hex-encoded data)
  • Verifying the fallback %v formatting for unknown types

These are exercised in the connector's actual use with BSON data.

🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.

In `@connectors/firestore/connector_test.go` around lines 104 - 125, Update
TestValueToString to include BSON and error scenarios: add table entries that
pass a bson.ObjectId (expecting its Hex() string) and a bson.Binary (expecting
hex-encoded data) to valueToString and assert the returned strings, add a case
with an unknown/complex type (e.g., a custom struct) to assert the fallback
fmt.Sprintf("%v") behavior, and add at least one input that should cause
valueToString to return an error and assert.Error; reference the existing
TestValueToString and valueToString function names when locating where to add
these cases.
🤖 Prompt for all review comments with AI agents
Verify each finding against the current code and only fix it if needed.

Inline comments:
In `@connectors/firestore/connector_integration_test.go`:
- Line 199: The test currently ignores the error returned from encodeJSON
(encoded, _ := encodeJSON(doc)), which can hide setup failures; update the test
to check the error and fail fast—e.g., replace the ignored error with error
handling using the test helper (t.Fatalf or require.NoError) to assert
encodeJSON(doc) returns no error and only then use the encoded value; reference
the encodeJSON call and ensure the test fails immediately if encoding fails.
- Line 180: The deferred type assertion connector.(interface{ Teardown()
}).Teardown() can panic if connector lacks that exact Teardown() method; change
it to perform a safe comma‑ok assertion inside the deferred function (e.g.,
defer func() { if td, ok := connector.(interface{ Teardown() }); ok {
td.Teardown() } }) so Teardown is only called when implemented and the test
won’t panic unexpectedly; optionally log or t.Log when Teardown is absent for
visibility.

In `@go.mod`:
- Line 45: Update the grpc dependency entry "google.golang.org/grpc v1.76.0" in
go.mod to v1.80.0 and ensure the module graph is refreshed (e.g., run go get
google.golang.org/grpc@v1.80.0 and go mod tidy) so the fix for CVE-2026-33186 is
applied; afterwards run the test suite/build to verify no regressions.

---

Nitpick comments:
In `@connectors/firestore/connector_test.go`:
- Around line 104-125: Update TestValueToString to include BSON and error
scenarios: add table entries that pass a bson.ObjectId (expecting its Hex()
string) and a bson.Binary (expecting hex-encoded data) to valueToString and
assert the returned strings, add a case with an unknown/complex type (e.g., a
custom struct) to assert the fallback fmt.Sprintf("%v") behavior, and add at
least one input that should cause valueToString to return an error and
assert.Error; reference the existing TestValueToString and valueToString
function names when locating where to add these cases.

In `@connectors/firestore/connector.go`:
- Around line 91-93: The code silently clamps invalid settings.BatchSize to
DefaultBatchSize; update the if block that checks settings.BatchSize to emit a
warning before changing the value so users know their configured value was
adjusted—e.g., log a message referencing the original settings.BatchSize and
DefaultBatchSize (use the connector's existing logger, e.g., logger.Warnf or
log.Printf) then set settings.BatchSize = DefaultBatchSize. Ensure you reference
settings.BatchSize and DefaultBatchSize in the warning so it's clear which
values were changed.
- Around line 322-325: The current switch cases for numeric types (case
int/int32/int64 and case uint/uint32/uint64) can cause precision loss for values
> JS_SAFE_INT (2^53-1); update the handling so uint64 (and unsigned types
promoted to uint64) are detected and for values > 9007199254740991 you return
their decimal representation as a string (use strconv.FormatUint(val, 10)),
otherwise return the numeric formatting; specifically modify the uint64 handling
branch (and any code paths that cast to uint64) to compare against the JS max
safe integer and return a string for large values to avoid downstream JSON
precision loss.
- Around line 199-220: In writeBatch, include context about which item failed by
changing the loop to capture the item index and include it in the returned error
when extractDocumentIDAndData fails; update the error returned from
extractDocumentIDAndData inside writeBatch (and any subsequent errors) to wrap
the original error with fmt.Errorf including the index (and optionally a short
snippet or hex of raw) and the collectionName so it's straightforward to
identify the problematic document when debugging; reference conn.writeBatch and
extractDocumentIDAndData to locate and modify the error wrapping logic.
🪄 Autofix (Beta)

Fix all unresolved CodeRabbit comments on this PR:

  • Push a commit to this branch (recommended)
  • Create a new PR with the fixes

ℹ️ Review info
⚙️ Run configuration

Configuration used: defaults

Review profile: CHILL

Plan: Pro

Run ID: 6ee06f1d-1eb1-4af4-88df-61acc4fd43a7

📥 Commits

Reviewing files that changed from the base of the PR and between 0a071d9 and c2a59b6.

⛔ Files ignored due to path filters (1)
  • go.sum is excluded by !**/*.sum
📒 Files selected for processing (5)
  • connectors/firestore/connector.go
  • connectors/firestore/connector_integration_test.go
  • connectors/firestore/connector_test.go
  • go.mod
  • internal/app/options/connectorflags.go

BatchSize: 100,
})
assert.NoError(t, err)
defer connector.(interface{ Teardown() }).Teardown()

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

⚠️ Potential issue | 🟡 Minor

Type assertion may panic if Teardown method signature changes.

The type assertion connector.(interface{ Teardown() }).Teardown() will panic at runtime if the connector doesn't implement this exact interface. Consider using a safe type assertion with the comma-ok idiom.

Proposed safer cleanup
-	defer connector.(interface{ Teardown() }).Teardown()
+	defer func() {
+		if t, ok := connector.(interface{ Teardown() }); ok {
+			t.Teardown()
+		}
+	}()
📝 Committable suggestion

‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.

Suggested change
defer connector.(interface{ Teardown() }).Teardown()
defer func() {
if t, ok := connector.(interface{ Teardown() }); ok {
t.Teardown()
}
}()
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.

In `@connectors/firestore/connector_integration_test.go` at line 180, The deferred
type assertion connector.(interface{ Teardown() }).Teardown() can panic if
connector lacks that exact Teardown() method; change it to perform a safe
comma‑ok assertion inside the deferred function (e.g., defer func() { if td, ok
:= connector.(interface{ Teardown() }); ok { td.Teardown() } }) so Teardown is
only called when implemented and the test won’t panic unexpectedly; optionally
log or t.Log when Teardown is absent for visibility.

"id": i, // JSON_ID uses "id" field
"value": i * 10,
}
encoded, _ := encodeJSON(doc)

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

⚠️ Potential issue | 🟡 Minor

Ignored error from encodeJSON could mask test setup failures.

If JSON encoding fails, the test would proceed with nil/empty data, potentially causing confusing failures later. Consider asserting no error.

Proposed fix
-		encoded, _ := encodeJSON(doc)
+		encoded, err := encodeJSON(doc)
+		assert.NoError(t, err)
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.

In `@connectors/firestore/connector_integration_test.go` at line 199, The test
currently ignores the error returned from encodeJSON (encoded, _ :=
encodeJSON(doc)), which can hide setup failures; update the test to check the
error and fail fast—e.g., replace the ignored error with error handling using
the test helper (t.Fatalf or require.NoError) to assert encodeJSON(doc) returns
no error and only then use the encoded value; reference the encodeJSON call and
ensure the test fails immediately if encoding fails.

Comment thread go.mod
google.golang.org/grpc v1.73.0
google.golang.org/protobuf v1.36.6
golang.org/x/time v0.14.0
google.golang.org/grpc v1.76.0

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

⚠️ Potential issue | 🔴 Critical

🧩 Analysis chain

🌐 Web query:

What is the latest patched version of google.golang.org/grpc that fixes the authorization bypass vulnerability GHSA-p77j-4mvh-x3m3?

💡 Result:

The latest patched version of google.golang.org/grpc that fixes the authorization bypass vulnerability GHSA-p77j-4mvh-x3m3 (CVE-2026-33186) is v1.80.0, released on 2026-04-01. This version includes all prior fixes, including the specific patch in v1.79.3 released on 2026-03-17.

Citations:


Upgrade google.golang.org/grpc to v1.80.0 to fix a critical authorization bypass vulnerability.

gRPC v1.76.0 contains GHSA-p77j-4mvh-x3m3 (CVE-2026-33186), a critical authorization bypass vulnerability caused by missing leading slash validation in the :path pseudo-header. This allows attackers to bypass authorization checks. Upgrade to v1.80.0 or later to resolve this.

🧰 Tools
🪛 OSV Scanner (2.3.5)

[CRITICAL] 45-45: google.golang.org/grpc 1.76.0: Authorization bypass in gRPC-Go via missing leading slash in :path in google.golang.org/grpc

(GO-2026-4762)


[CRITICAL] 45-45: google.golang.org/grpc 1.76.0: gRPC-Go has an authorization bypass via missing leading slash in :path

(GHSA-p77j-4mvh-x3m3)

🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.

In `@go.mod` at line 45, Update the grpc dependency entry "google.golang.org/grpc
v1.76.0" in go.mod to v1.80.0 and ensure the module graph is refreshed (e.g.,
run go get google.golang.org/grpc@v1.80.0 and go mod tidy) so the fix for
CVE-2026-33186 is applied; afterwards run the test suite/build to verify no
regressions.

@coderabbitai coderabbitai Bot left a comment

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Actionable comments posted: 2

🤖 Prompt for all review comments with AI agents
Verify each finding against the current code and only fix it if needed.

Inline comments:
In `@connectors/firestore/connector.go`:
- Around line 250-282: The switch currently treats UPDATE_TYPE_PARTIAL_UPDATE as
a full upsert; add an explicit case for
adiomv1.UpdateType_UPDATE_TYPE_PARTIAL_UPDATE in the switch and handle it by
converting update.GetData() via rawToMap (same as other branches), deleting
idKey, and calling the incremental update method on the batch writer (e.g.,
bw.Update(docRef, partialData)) with the same bw.End() and error wrapping
pattern; if the batch writer does not support partial updates, return a clear
error instead of falling through to bw.Set to avoid overwriting full documents.
- Around line 435-451: The valueToString function currently falls back to
fmt.Sprintf("%v", val) which silently converts complex types (maps, slices,
structs) into unstable IDs; change valueToString to only accept explicit types:
string, bson.ObjectID (Hex), integer/unsigned/float families (formatted),
bson.Binary (hex of Data), []byte (hex), and any type implementing fmt.Stringer
(use .String()); for any other type return a descriptive error like "unsupported
id type: %T" instead of the %v fallback so callers can handle invalid ID types
instead of producing unstable document IDs.
🪄 Autofix (Beta)

Fix all unresolved CodeRabbit comments on this PR:

  • Push a commit to this branch (recommended)
  • Create a new PR with the fixes

ℹ️ Review info
⚙️ Run configuration

Configuration used: defaults

Review profile: CHILL

Plan: Pro

Run ID: 0471fea1-61f9-45f2-96b2-7dd758da67b8

📥 Commits

Reviewing files that changed from the base of the PR and between c2a59b6 and e232934.

📒 Files selected for processing (3)
  • connectors/firestore/bson_conversion_test.go
  • connectors/firestore/connector.go
  • connectors/firestore/connector_integration_test.go
🚧 Files skipped from review as they are similar to previous changes (1)
  • connectors/firestore/connector_integration_test.go

Comment on lines +250 to +282
switch update.GetType() {
case adiomv1.UpdateType_UPDATE_TYPE_DELETE:
job, err = bw.Delete(docRef)
if err != nil {
bw.End()
return fmt.Errorf("failed to queue delete operation: %w", err)
}
case adiomv1.UpdateType_UPDATE_TYPE_UPDATE, adiomv1.UpdateType_UPDATE_TYPE_INSERT:
docData, err := rawToMap(update.GetData(), dataType)
if err != nil {
bw.End()
return fmt.Errorf("failed to convert update data: %w", err)
}
delete(docData, idKey)
job, err = bw.Set(docRef, docData)
if err != nil {
bw.End()
return fmt.Errorf("failed to queue set operation: %w", err)
}
default:
slog.Warn("unknown update type, treating as upsert", "type", update.GetType())
docData, err := rawToMap(update.GetData(), dataType)
if err != nil {
bw.End()
return fmt.Errorf("failed to convert update data: %w", err)
}
delete(docData, idKey)
job, err = bw.Set(docRef, docData)
if err != nil {
bw.End()
return fmt.Errorf("failed to queue set operation: %w", err)
}
}

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

⚠️ Potential issue | 🔴 Critical

UPDATE_TYPE_PARTIAL_UPDATE currently falls into full upsert behavior.

This can overwrite full documents with partial payloads and cause data loss. Handle partial updates explicitly (or reject them) instead of routing through the default upsert path.

Proposed safe handling
 		switch update.GetType() {
 		case adiomv1.UpdateType_UPDATE_TYPE_DELETE:
 			job, err = bw.Delete(docRef)
@@
 		case adiomv1.UpdateType_UPDATE_TYPE_UPDATE, adiomv1.UpdateType_UPDATE_TYPE_INSERT:
 			docData, err := rawToMap(update.GetData(), dataType)
 			if err != nil {
 				bw.End()
 				return fmt.Errorf("failed to convert update data: %w", err)
 			}
 			delete(docData, idKey)
 			job, err = bw.Set(docRef, docData)
 			if err != nil {
 				bw.End()
 				return fmt.Errorf("failed to queue set operation: %w", err)
 			}
+		case adiomv1.UpdateType_UPDATE_TYPE_PARTIAL_UPDATE:
+			bw.End()
+			return fmt.Errorf("partial updates are not supported by firestore sink")
 		default:
-			slog.Warn("unknown update type, treating as upsert", "type", update.GetType())
-			docData, err := rawToMap(update.GetData(), dataType)
-			if err != nil {
-				bw.End()
-				return fmt.Errorf("failed to convert update data: %w", err)
-			}
-			delete(docData, idKey)
-			job, err = bw.Set(docRef, docData)
-			if err != nil {
-				bw.End()
-				return fmt.Errorf("failed to queue set operation: %w", err)
-			}
+			bw.End()
+			return fmt.Errorf("unsupported update type: %v", update.GetType())
 		}
📝 Committable suggestion

‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.

Suggested change
switch update.GetType() {
case adiomv1.UpdateType_UPDATE_TYPE_DELETE:
job, err = bw.Delete(docRef)
if err != nil {
bw.End()
return fmt.Errorf("failed to queue delete operation: %w", err)
}
case adiomv1.UpdateType_UPDATE_TYPE_UPDATE, adiomv1.UpdateType_UPDATE_TYPE_INSERT:
docData, err := rawToMap(update.GetData(), dataType)
if err != nil {
bw.End()
return fmt.Errorf("failed to convert update data: %w", err)
}
delete(docData, idKey)
job, err = bw.Set(docRef, docData)
if err != nil {
bw.End()
return fmt.Errorf("failed to queue set operation: %w", err)
}
default:
slog.Warn("unknown update type, treating as upsert", "type", update.GetType())
docData, err := rawToMap(update.GetData(), dataType)
if err != nil {
bw.End()
return fmt.Errorf("failed to convert update data: %w", err)
}
delete(docData, idKey)
job, err = bw.Set(docRef, docData)
if err != nil {
bw.End()
return fmt.Errorf("failed to queue set operation: %w", err)
}
}
switch update.GetType() {
case adiomv1.UpdateType_UPDATE_TYPE_DELETE:
job, err = bw.Delete(docRef)
if err != nil {
bw.End()
return fmt.Errorf("failed to queue delete operation: %w", err)
}
case adiomv1.UpdateType_UPDATE_TYPE_UPDATE, adiomv1.UpdateType_UPDATE_TYPE_INSERT:
docData, err := rawToMap(update.GetData(), dataType)
if err != nil {
bw.End()
return fmt.Errorf("failed to convert update data: %w", err)
}
delete(docData, idKey)
job, err = bw.Set(docRef, docData)
if err != nil {
bw.End()
return fmt.Errorf("failed to queue set operation: %w", err)
}
case adiomv1.UpdateType_UPDATE_TYPE_PARTIAL_UPDATE:
bw.End()
return fmt.Errorf("partial updates are not supported by firestore sink")
default:
bw.End()
return fmt.Errorf("unsupported update type: %v", update.GetType())
}
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.

In `@connectors/firestore/connector.go` around lines 250 - 282, The switch
currently treats UPDATE_TYPE_PARTIAL_UPDATE as a full upsert; add an explicit
case for adiomv1.UpdateType_UPDATE_TYPE_PARTIAL_UPDATE in the switch and handle
it by converting update.GetData() via rawToMap (same as other branches),
deleting idKey, and calling the incremental update method on the batch writer
(e.g., bw.Update(docRef, partialData)) with the same bw.End() and error wrapping
pattern; if the batch writer does not support partial updates, return a clear
error instead of falling through to bw.Set to avoid overwriting full documents.

Comment on lines +435 to +451
func valueToString(v any) (string, error) {
switch val := v.(type) {
case string:
return val, nil
case bson.ObjectID:
return val.Hex(), nil
case int, int32, int64:
return fmt.Sprintf("%d", val), nil
case uint, uint32, uint64:
return fmt.Sprintf("%d", val), nil
case float32, float64:
return fmt.Sprintf("%v", val), nil
case bson.Binary:
return fmt.Sprintf("%x", val.Data), nil
default:
return fmt.Sprintf("%v", val), nil
}

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

⚠️ Potential issue | 🟠 Major

valueToString silently accepts unsupported ID types.

The default %v fallback converts maps/slices/structs into ad-hoc strings, which can create unstable/colliding document IDs. This should return an error for unsupported types.

Proposed stricter ID conversion
 func valueToString(v any) (string, error) {
 	switch val := v.(type) {
 	case string:
 		return val, nil
 	case bson.ObjectID:
 		return val.Hex(), nil
 	case int, int32, int64:
 		return fmt.Sprintf("%d", val), nil
 	case uint, uint32, uint64:
 		return fmt.Sprintf("%d", val), nil
 	case float32, float64:
 		return fmt.Sprintf("%v", val), nil
 	case bson.Binary:
 		return fmt.Sprintf("%x", val.Data), nil
 	default:
-		return fmt.Sprintf("%v", val), nil
+		return "", fmt.Errorf("unsupported document ID type: %T", v)
 	}
 }
📝 Committable suggestion

‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.

Suggested change
func valueToString(v any) (string, error) {
switch val := v.(type) {
case string:
return val, nil
case bson.ObjectID:
return val.Hex(), nil
case int, int32, int64:
return fmt.Sprintf("%d", val), nil
case uint, uint32, uint64:
return fmt.Sprintf("%d", val), nil
case float32, float64:
return fmt.Sprintf("%v", val), nil
case bson.Binary:
return fmt.Sprintf("%x", val.Data), nil
default:
return fmt.Sprintf("%v", val), nil
}
func valueToString(v any) (string, error) {
switch val := v.(type) {
case string:
return val, nil
case bson.ObjectID:
return val.Hex(), nil
case int, int32, int64:
return fmt.Sprintf("%d", val), nil
case uint, uint32, uint64:
return fmt.Sprintf("%d", val), nil
case float32, float64:
return fmt.Sprintf("%v", val), nil
case bson.Binary:
return fmt.Sprintf("%x", val.Data), nil
default:
return "", fmt.Errorf("unsupported document ID type: %T", v)
}
}
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.

In `@connectors/firestore/connector.go` around lines 435 - 451, The valueToString
function currently falls back to fmt.Sprintf("%v", val) which silently converts
complex types (maps, slices, structs) into unstable IDs; change valueToString to
only accept explicit types: string, bson.ObjectID (Hex), integer/unsigned/float
families (formatted), bson.Binary (hex of Data), []byte (hex), and any type
implementing fmt.Stringer (use .String()); for any other type return a
descriptive error like "unsupported id type: %T" instead of the %v fallback so
callers can handle invalid ID types instead of producing unstable document IDs.

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

None yet

Projects

None yet

Development

Successfully merging this pull request may close these issues.

1 participant